package com.huonan.component.kafka.config;

import lombok.Getter;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.IsolationLevel;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.annotation.EnableKafka;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.config.KafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;
import org.springframework.kafka.listener.ContainerProperties;

import java.util.HashMap;
import java.util.Locale;
import java.util.Map;

/**
 *
 */
@Configuration
@EnableKafka
@Getter
public class KafkaConsumerConfig {
    @Value("${kafka.server.list:10.11.110.78:9092}")
    private String bootstrapServers;
    @Value("${spring.kafka.consumer.max-poll-records:1000}")
    private Integer maxPollRecords;
    // 批次消息处理时间。调用 poll() 之间的最大延迟，超时到期，则认为消费者失败，组将重新平衡。
    @Value("${spring.kafka.consumer.max-poll-interval-ms:600000}")
    private Integer maxPollInterval;
    @Value("${spring.kafka.listener.container.poll-timeout:20000}")
    private Integer pollTimeout;
    @Value("${spring.kafka.consumer.fetch-min-bytes:512000}") // 500KB
    private Integer fetchMinBytes;
    @Value("${spring.kafka.consumer.fetch-max-bytes:104857600}") // 100MB
    private Integer fetchMaxBytes;
    @Value("${spring.kafka.consumer.max-partition-fetch-bytes:12582912}") // 12MB
    private Integer maxPartitionFetchBytes;
    @Value("${spring.kafka.consumer.auto-offset-reset-config:latest}")
    private String autoOffsetResetConfig;

    @Bean
    public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, String>> batchFactory() {
        ConcurrentKafkaListenerContainerFactory<String, String> factory =
                new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory());
        factory.setBatchListener(true);
        // 如果没有达到fetch.min.bytes(1KB)，最长等待时间。消费者等待记录的最长时间。
        factory.getContainerProperties().setPollTimeout((long) pollTimeout);
        factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL);
        return factory;
    }

    @Bean
    public ConsumerFactory<String, String> consumerFactory() {
        return new DefaultKafkaConsumerFactory<>(consumerConfigs());
    }

    @Bean
    public Map<String, Object> consumerConfigs() {
        Map<String, Object> props = new HashMap<>();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, maxPollRecords);
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, autoOffsetResetConfig);
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
        props.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG, IsolationLevel.READ_COMMITTED.toString().toLowerCase(Locale.ROOT));
        props.put(ConsumerConfig.FETCH_MIN_BYTES_CONFIG, fetchMinBytes);
        props.put(ConsumerConfig.FETCH_MAX_BYTES_CONFIG, fetchMaxBytes);
        props.put(ConsumerConfig.FETCH_MAX_BYTES_CONFIG, fetchMaxBytes);
        props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, 10 * 1000);
        // 如果没有足够的数据立即满足 fetch.min.bytes 给出的要求，服务器将在响应 fetch 请求之前阻塞的最长时间。
        props.put(ConsumerConfig.FETCH_MAX_WAIT_MS_CONFIG, pollTimeout); // 同pollTimeout
        // Member sending LeaveGroup request to coordinator 的时间只与这个参数有关
        props.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, maxPollInterval);
        // 控制客户端等待请求响应的最长时间
        props.put(ConsumerConfig.REQUEST_TIMEOUT_MS_CONFIG, 600 * 1000);
        // 默认为10秒,客户端故障的超时
        props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 200 * 1000);
        return props;
    }
}
